package com.xj.hadoop.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.PageFilter;

import java.io.IOException;

public class DataBase<T> {
	private static Configuration conf = null;
	private static HTablePool pool = null;
	private HTableInterface table=null;
	static {
		conf = HBaseConfiguration.create();
		//提高RPC通信时长
		conf.setLong("hbase.rpc.timeout", 600000);
		//设置Scan缓存
		conf.setLong("hbase.client.scanner.caching", 1000);
		conf.set("hbase.zookeeper.quorum", "服务器地址");
		pool=new HTablePool(conf, 1000);
	}

	public boolean createTable(String tableName,String [] column) throws Exception {
		HBaseAdmin admin = new HBaseAdmin(conf);
		if(admin.tableExists(tableName)){
			throw new Exception(tableName+" 已经存在.");
		}else if(column==null){
			throw new Exception("column =null. 列至少有一个。");
		}else{
			HTableDescriptor tableDesc = new HTableDescriptor(tableName);
			for (int i = 0; i < column.length; i++) {
				tableDesc.addFamily(new HColumnDescriptor(column[i]));
	            admin.createTable(tableDesc);
			}
            return true;
		}
	}

	public T selectOne(String tablename,String rowKey) throws IOException {
		//table = new HTable(conf, tablename);
		table=pool.getTable(tablename);
		Get g = new Get(rowKey.getBytes());
		Result rs = table.get(g);
		for (KeyValue kv : rs.raw()) {
			System.out.print(new String(kv.getRow()) + "  ");
			System.out.print(new String(kv.getFamily()) + ":");
			System.out.print(new String(kv.getQualifier()) + "  ");
			System.out.print(kv.getTimestamp() + "  ");
			System.out.println(new String(kv.getValue()));
		}
		return null;
	}
	public T selectList(String tablename,String [] col,String startRow,int limit) throws IOException{
		//table = new HTable(conf, tablename);
		table=pool.getTable(tablename);
		Scan scan=new Scan();
		if(col!=null){
			for (int i = 0; i < col.length; i++) {
				String [] col_=col[i].split(":");
				if(col_.length==2){
					scan.addColumn(col_[0].getBytes(), col_[1].getBytes());
				}else{
					scan.addFamily(col_[i].getBytes());
				}
			}
		}
		if(startRow!=null&&!startRow.equals("")){
			scan.setStartRow(startRow.getBytes());
		}
		if(limit>0){
			Filter f=new PageFilter(limit);
			scan.setFilter(f);
			scan.setCaching(limit);
		}
		ResultScanner rs=table.getScanner(scan);
		Result r=null;
		while((r=rs.next())!=null){
			for (KeyValue kv : r.raw()) {
				System.out.print(new String(kv.getRow()) + "  ");
				System.out.print(new String(kv.getFamily()) + ":");
				System.out.print(new String(kv.getQualifier()) + "  ");
				System.out.print(kv.getTimestamp() + "  ");
				System.out.println(new String(kv.getValue()));
			}
		}
		rs.close();
		//table.close();//不要关闭
		return null;
	}
	public void add(String tableName) throws IOException{
		HTable table = new HTable(conf, tableName);
		Put put =new Put("key".getBytes());//主键
		put.add("列簇".getBytes(), "列名".getBytes(), "value".getBytes());
		table.put(put);
	}
	public void delete(String tableName,String key,String col) throws IOException {
		HTable table = new HTable(conf, tableName);
		//table.setAutoFlush(false);
		Delete del=new Delete(key.getBytes());
		del.deleteColumn(col.split(":")[0].getBytes(), col.split(":")[1].getBytes());
		//del.deleteFamily(family); 列 前缀
		table.delete(del);
	}
	public void deleteAll(String tableName,String key) throws IOException{
		HTable table = new HTable(conf, tableName);
		Delete del=new Delete(key.getBytes());
		table.delete(del);
	}
	public void count(String tableName) throws Throwable{
		AggregationClient ac = new AggregationClient(conf);
		Scan scan = new Scan();
		scan.addFamily("stu".getBytes());
		long rowCount = ac.rowCount(tableName.getBytes(), null, scan);
		System.out.println(rowCount);
	}
	public static void main(String[] args) throws Exception {
		DataBase db=new DataBase();
		String [] col={"teacher"};
		db.createTable("person2",col);
		
		//db.selectOne("person2", "123");
		//db.add("person2");
		//db.delete("person2", "123", "teacher:1");
		//db.deleteAll("person2", "123");
		//String []col={"student:name"};
		//db.selectList("person", col, "2", 2);
		/*try {
			db.count("person2");
		} catch (Throwable e) {
			e.printStackTrace();
		}*/
	}
}
